package com.mango.ch07;

/**
 * 理解代码后，默写的实现代码
 */
import java.io.IOException;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFlatMapFunction;

import com.mango.HDFSTools.HDFSAPI;

import scala.Tuple2;
import scala.Tuple3;

public class myTest {
	static Path inputPath = new Path("/MBA_Job/input");// input
	static Path outPath = new Path("/MBA_Job/output");

	@SuppressWarnings("serial")
	public static void main(String[] args) {
		// TODO Auto-generated method stub
		// 创建输入文件夹
		Configuration conf = new Configuration();
		try {
			HDFSAPI hdfs = new HDFSAPI(conf);
			// hdfs.createDirectory(transactions);
			// 先删除已经有的输出文件夹
			hdfs.delDirectory(outPath);
			// hdfs.orpOver();
		} catch (IOException e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
			System.out.println("----------文件操作失败");
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (URISyntaxException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		SparkConf sconf = new SparkConf();
		sconf.setAppName("Spark_MBAJob");
		// 建立一个快速串化器
		sconf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");
		// 现在默认为32mb的缓冲区而不是64mb
		sconf.set("spark.Kryoserializer.buffer.mb", "32");
		JavaSparkContext jsc = new JavaSparkContext(sconf);
		JavaRDD<String> line = jsc.textFile(inputPath.toString());
		// 生成频繁模式
		JavaPairRDD<List<String>, Integer> allPatterns = line
				.flatMapToPair(new PairFlatMapFunction<String, List<String>, Integer>() {

					@Override
					public Iterator<Tuple2<List<String>, Integer>> call(String value) throws Exception {
						// TODO Auto-generated method stub
						List<Tuple2<List<String>, Integer>> result = new ArrayList<>();
						List<String> elements = toList(value.trim());
						List<List<String>> combinations = Combination.findSortedCombinatinos(elements);
						for (List<String> combination : combinations) {
							if (combination.size() > 0)
								result.add(new Tuple2<List<String>, Integer>(combination, 1));
						}
						return result.iterator();
					}
				});
		// 规约频繁模式
		JavaPairRDD<List<String>, Integer> uniquePatterns = allPatterns
				.reduceByKey(new Function2<Integer, Integer, Integer>() {

					@Override
					public Integer call(Integer arg0, Integer arg1) throws Exception {
						// TODO Auto-generated method stub
						return arg0 + arg1;
					}
				});
		// 生成子模式(频繁模式的列表-1 之后的结果，如：([a, d],1)生成
		// ([a, d],(null,1))
		// ([d],([a, d],1))
		// ([a],([a, d],1))
		JavaPairRDD<List<String>, Tuple2<List<String>, Integer>> subPatterns = uniquePatterns.flatMapToPair(
				new PairFlatMapFunction<Tuple2<List<String>, Integer>, List<String>, Tuple2<List<String>, Integer>>() {

					@Override
					public Iterator<Tuple2<List<String>, Tuple2<List<String>, Integer>>> call(
							Tuple2<List<String>, Integer> patterns) throws Exception {
						// TODO Auto-generated method stub
						List<Tuple2<List<String>, Tuple2<List<String>, Integer>>> result = new ArrayList<>();
						List<String> oriList = patterns._1;
						Integer requency = patterns._2;
						result.add(new Tuple2<List<String>, Tuple2<List<String>, Integer>>(oriList,
								new Tuple2<List<String>, Integer>(null, requency)));
						if (oriList.size() == 1)// 如果频繁模式的列表只有一个元素，则没有子模式
							return result.iterator();
						for (int i = 0; i < oriList.size(); i++) {
							List<String> list = removeOneItem(oriList, i);
							result.add(new Tuple2<List<String>, Tuple2<List<String>, Integer>>(list,
									new Tuple2<List<String>, Integer>(oriList, requency)));
						}
						return result.iterator();
					}
				});
		// 规约子模式
		JavaPairRDD<List<String>, Iterable<Tuple2<List<String>, Integer>>> uniqueSubPatterns = subPatterns.groupByKey();
		// 生成关联规则
		JavaRDD<List<Tuple3<List<String>, List<String>, Double>>> finaResult = uniqueSubPatterns.map(
				new Function<Tuple2<List<String>, Iterable<Tuple2<List<String>, Integer>>>, List<Tuple3<List<String>, List<String>, Double>>>() {

					@Override
					public List<Tuple3<List<String>, List<String>, Double>> call(
							Tuple2<List<String>, Iterable<Tuple2<List<String>, Integer>>> patterns) throws Exception {
						// TODO Auto-generated method stub
						List<Tuple3<List<String>, List<String>, Double>> result = new ArrayList<Tuple3<List<String>, List<String>, Double>>();
						List<String> fromList = patterns._1;// 代表X---关联--->Y中的 X
						Iterable<Tuple2<List<String>, Integer>> to = patterns._2;//// 代表X---关联--->Y中的 Y(现在是个list集)
						List<Tuple2<List<String>, Integer>> toList = new ArrayList<>();// 找到(null,2)项，他记录了X总共出现了多少次
						Tuple2<List<String>, Integer> fromCount = null;
						for (Tuple2<List<String>, Integer> t2 : to) {
							// 将表示Y的Iterable中的元素 遍历添加到新的临时list中
							if (t2._1 == null) {
								fromCount = t2;
							} else
								toList.add(t2);
						}
						if (toList.isEmpty()) {
							// 如果临时list是空的表示，x->y 模式中，x没有关联任何对象
							return result;
						}
						for (Tuple2<List<String>, Integer> t2 : toList) {
							// 否则遍历临时list，计算置信度，添加到resul中
							double confidence = (double) t2._2() / (double) fromCount._2;
							List<String> t2List = new ArrayList<>(t2._1);// 如t2._1为 [a, b, d]
							t2List.removeAll(fromList);
							result.add(new Tuple3<List<String>, List<String>, Double>(fromList, t2List, confidence));
						}
						return result;
					}
				});
		finaResult.saveAsTextFile(outPath.toString() + "/final");
		jsc.close();
		System.exit(0);
	}

	/**
	 * 将一行字符串切割后转换为一个list
	 * 
	 * @param transaction
	 * @return
	 */
	public static List<String> toList(String transaction) {
		String[] items = transaction.trim().split(",");
		List<String> list = new ArrayList<>();
		for (String string : items) {
			list.add(string);
		}
		return list;
	}

	/**
	 * 移除list中下标为i的元素
	 * 
	 * @param list
	 * @param i
	 * @return
	 */
	public static List<String> removeOneItem(List<String> list, int i) {
		if ((list == null) || (list.isEmpty())) {
			return list;
		}
		if ((i < 0) || i > list.size()) {
			return list;
		}
		List<String> cloned = new ArrayList<>(list);
		cloned.remove(i);
		return cloned;
	}

}
